package com.luoqifei.spark;

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;

public class SparkStreamCase {
    public static void main(String[] args) throws InterruptedException {
        SparkConf sparkConf = new SparkConf().setMaster("local").setAppName("test");
        JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, Durations.seconds(1));
        JavaReceiverInputDStream<String> lines= jssc.socketTextStream("localhost",8980);

        JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());

        JavaPairDStream<String,Integer> pairs = words.mapToPair(s -> new Tuple2<String,Integer>(s,1));
        JavaPairDStream<String,Integer> wordCounts = pairs.reduceByKey((i1,i2) -> i1+i2);
        wordCounts.print();
        jssc.start();
        jssc.awaitTermination();
    }
}
